﻿using System;
using System.Linq;
using System.Collections.Generic;
using System.Text;
using System.IO;
using MPI;

namespace Dwarf.Managed
{
    /// <summary>Solve the MapReduce problem with MPI parallelization</summary>
    ///
    class Solver
    {                                                       // MPI variables:
        MPI.Environment mpienv;                             // enviroment
        MPI.Intracommunicator incomm;                       // intracommunicator
        int rank;                                           // rank of the process
        int commsize;                                       // count of processes
        int root = 0;                                       // index of root process. By consideration, equal to 0

        private Configurator dwarfConfigurator;             // processed settings from command line
        private string mainContent;                         // content for parsing
        private IDictionary<String, int> stringTotal;  // final result Dictionary "word" - "count of entries"
        private string[] mpiBounds;                         // content partitioning for mpi-processors

        /// <summary>Construct the new solver</summary>
        /// <param name="dwarfConfigurator">Processed command line arguments</param>
        /// <param name="args">Command Line arguments</param>
        /// 
        public Solver(ref Configurator dwarfConfigurator, string[] args)
        {
            mpienv = new MPI.Environment(ref args);      // MPI Initialization
            rank = MPI.Communicator.world.Rank;          // rank of the processors
            commsize = MPI.Communicator.world.Size;      // count of processes
            incomm = MPI.Intracommunicator.world;

            if (rank == root)
            {
                this.dwarfConfigurator = dwarfConfigurator;
                dwarfConfigurator.WriteSettings();               // write settings to console and profile file
                mainContent = dwarfConfigurator.GetData();

                mpiBounds = new string[commsize];
                InitParallelization(commsize);
            }
            incomm.Broadcast<string>(ref mpiBounds, root);      // send the array of data to all proccessors
        }

        /// <summary>MPI method MapReduce for each 'rank' - processor</summary>
        ///
        public void Solve()
        {
            var currentData = new MapReduceBaseMethod(mpiBounds[rank]);                     // content data for each process is defined by its rank and rank-position in mpiBounds array
            currentData.FunctionMap();                                                      // start method Map for all processes
            currentData.FunctionReduce();                                                   // start method Reduce for all processes
            var op = new ReductionOperation<Dictionary<string, int>>(SumThreadResult);      // custom operation for all threads result reduce
            stringTotal = incomm.Reduce<Dictionary<string, int>>(currentData.GetStringTotal(), op, root); // mpi-reduce
            if (rank == root)
                stringTotal = new SortedDictionary<string, int>(stringTotal);
        }

        /// <summary>Init the configuration. Creating objects for parallel processing</summary>
        /// <param name="threadCount">Count of threads used for parallelization</param>
        ///
        private void InitParallelization(int threadCount)
        {
            int contentSize = mainContent.Length;
            int interval = 1 + contentSize / threadCount;   // partitioning the String content (same parts for all threads)
            int leftBoard = 0;
            int part = 0;
            while (part < threadCount - 1)
            {
                int certainInterval = interval;
                while (mainContent[leftBoard + certainInterval] != ' ') // search of middle-word space
                {
                    certainInterval++;
                }
                certainInterval++;
                mpiBounds[part] = mainContent.Substring(leftBoard, certainInterval); // create the contents for all parallel methods
                part++;
                leftBoard += certainInterval;
            }
            mpiBounds[threadCount - 1] = mainContent.Substring(leftBoard);
        }

        /// <summary>Summarize two dictionaries by values' summarizing</summary>
        /// <param name="dictionaryFirst">The first summand</param>
        /// <param name="dictionarySecond">The second summand</param>
        /// <returns>Sum of Dictionaries</returns>
        ///
        private Dictionary<string, int> SumThreadResult(
            Dictionary<string, int> dictionaryFirst,
            Dictionary<string, int> dictionarySecond)
        {
            var tmpStringTotal = new Dictionary<String, int>();            // dictionary for final result
            try
            {
                IEnumerable<String> stringDistinct = dictionaryFirst.Keys;
                stringDistinct = dictionarySecond.Keys.Union(stringDistinct); // union of unique strings of the entries
                foreach (string key in stringDistinct)
                {
                    int value = 0;
                    foreach (var pair in dictionaryFirst)
                    {
                        if (pair.Key == key) { value += pair.Value; break; }  // search of entries' counts for each unique string
                    }
                    foreach (var pair in dictionarySecond)
                    {
                        if (pair.Key == key) { value += pair.Value; break; }  // search of entries' counts for each unique string
                    }
                    tmpStringTotal.Add(key, value);
                }
            }
            catch (Exception e)
            {
                Console.WriteLine("Error: {0}", e.Message);
            }
            return tmpStringTotal;
        }

        /// <summary>Problem results output</summary>
        /// <param name="time">Spent time</param>
        ///
        public void Finish(double time)
        {
            if (rank == root)
            {
                dwarfConfigurator.Close(time, stringTotal);
            }
            mpienv.Dispose();
        }
    }
}
